package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.UserBehavior;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**
 * Created by Smexy on 2023/1/29
 * 
 *     统计每小时网站的PV
 *
 */
public class Demo1_PV
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //创建水印生成策略
        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy
            .<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTimestamp());
        
        //env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
        //读数据
        env.readTextFile("data/UserBehavior.csv")
           .map(new MapFunction<String, UserBehavior>()
           {
               @Override
               public UserBehavior map(String line) throws Exception {
                   String[] words = line.split(",");
                  return new UserBehavior(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       Integer.valueOf(words[2]),
                       words[3],
                       //时间戳必须是ms
                       Long.valueOf(words[4]) * 1000
                   );
               }
           })
           .assignTimestampsAndWatermarks(watermarkStrategy)
           //过滤出PV的数据
           .filter(ub -> "pv".equals(ub.getBehavior()))
           //汇总，一定是一个全局窗口。 全局窗口计算时，并行度是1
           .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
           .process(new ProcessAllWindowFunction<UserBehavior, String, TimeWindow>()
           {
               //全量聚合。 在窗口触发时只执行一次
               @Override
               public void process(Context context, Iterable<UserBehavior> elements, Collector<String> out) throws Exception {
                   int pv = MyUtil.parseList(elements).size();
                   out.collect(MyUtil.parseTime(context.window()) + " 的PV:"+pv);
               }
           })
           .print().setParallelism(1);


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
        
        
    }
}
